package com.aididi.config;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author ：xuzexiang
 * @description：    事务监听器
 * @date ： 2022/4/8 16:25
 */
public class TransactionListenerImpl implements TransactionListener {

    private AtomicInteger transactionIndex = new AtomicInteger(0);
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    // 发送消息时走 executeLocalTransaction 方法逻辑
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        String tags = message.getTags();
        if(StringUtils.contains(tags,"TagA")){
            //返回COMMIT_MESSAGE状态的消息会立即被消费者消费到。
            return LocalTransactionState.COMMIT_MESSAGE;
        }else if(StringUtils.contains(tags,"TagB")){
            //返回ROLLBACK_MESSAGE状态的消息会被丢弃
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }else{
            //返回UNKNOW状态的消息会等待Broker进行事务状态回查
            return LocalTransactionState.UNKNOW;
        }
    }

    // 事务状态回查时，执行 checkLocalTransaction 方法，回查触发需要一定时间
    // 在 Broker 配置文件中的参数 transactionMsgTimeout 可以配置多久之后被检查
    //当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于 transactionMsgTimeout 参数
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        String tags = messageExt.getTags();
        if(StringUtils.contains(tags,"TagC")){

            //过一段时间后，TagC的消息被消费者消费到
            return LocalTransactionState.COMMIT_MESSAGE;
        }else if(StringUtils.contains(tags,"TagD")){

            //TagD的消息也会在状态回查时被丢弃掉
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }else{

            //剩下TagE的消息会继续回查，回查15次依旧没有commit，则最终被丢弃
            //可以通过 Broker 配置文件的 transactionCheckMax参数来修改15次限制
            return LocalTransactionState.UNKNOW;
        }

    }
}
